package edu.csl.study.spark.basic

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
 * 通过IDEA直接提交到集群中。
 * 优势：IDEA作为Driver，可以直接调试集群运行。
 * 1.maven的package打包 StudySpark-1.0-SNAPSHOT.jar
 * 2.setJars 方法设置StudySpark-1.0-SNAPSHOT.jar的路径
 * 3.spark.driver.host 为IDEA所在IP地址，注意：要确保集群和本地的网络是通的才行。
 *   因为IDEA所在的服务器作为Driver，会启动Driver进程接收Executer进程汇报的任务执行状态。
 * 4.打包完直接run即可
 */
object Core_IDEA_Direct_To_Cluster {
  /**
   * 获取上下文
   *
   * @return
   */
  def newContext:SparkContext={

    val conf:SparkConf = new SparkConf().setAppName("WordCount").setMaster("spark://centos20:7077")
      .setJars(List(System.getProperty("user.dir") +"\\target\\StudySpark-1.0-SNAPSHOT.jar"))
      .setIfMissing("spark.driver.host","192.168.100.1")//
    val sc = new SparkContext(conf)

    sc.setLogLevel("warn")
    sc
  }
  def main(args: Array[String]): Unit = {

     println(System.getProperty("user.dir"))
      // 1.配置信息
     val sc = newContext


     val rdd2 = sc.parallelize(List(5, 6, 4, 3))
     val rdd3 = sc.makeRDD(List(1, 2, 3, 4))
     //求交集
     val rdd21 = rdd2.mapPartitionsWithIndex((index,iter) => {

      iter.map(x => {
        println("第"+index+"个分区数据： "+x )
        x * 10
      } )
    })
    println("---按分区mapPartitionsWithIndex--partitions.length = "+rdd2.partitions.length)
    //打印结果
    println("---按分区mapPartitionsWithIndex---结果："+rdd21.collect().mkString(","))
    println("---交集intersection---partitions.length = "+rdd2.partitions.length)
    println("---交集intersection-- -结果："+rdd2.intersection(rdd3).collect().mkString(","))
    //求并集
    println("---并集union---partitions.length = "+rdd3.partitions.length)
    println("---并集union---结果："+rdd2.union(rdd3).collect().mkString(","))


    val rdd4 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)))
    val rdd5 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("jim", 2)))

    /***
     * 分组cogroup: 在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD
     * 结果：
     * (jim,(CompactBuffer(),CompactBuffer(2)))
     * |(tom,(CompactBuffer(1, 2),CompactBuffer(1)))
     * |(jerry,(CompactBuffer(3),CompactBuffer(2)))
     * |(kitty,(CompactBuffer(2),CompactBuffer()))
     */
     Thread.sleep(60000);
    println("---cogroup---结果："+rdd4.cogroup(rdd5).collect().mkString("|"))
      //关闭上下文
      sc.stop()
  }


}
